package com.rocketmq.aliyun.consume;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @author zhangyan
 * @version $ Id: SequenceConsumer.java, v 0.1 2021/5/24 20:27 zhangyan Exp $
 */
public class SequenceConsumer {
    public static void main(String[] args) {

        /**
         * 创建Consumer，并开启消息轨迹。
         * 如果不想开启消息轨迹，可以按照如下方式创建：
         * DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("YOUR ORDER GROUP ID", getAclRPCHook(), null);
         */
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("groupSequence", true);

        /**
         * 设置使用接入方式为阿里云，在使用云上消息轨迹的时候，需要设置此项，如果不开启消息轨迹功能，则运行不设置此项。
         */
        consumer.setAccessChannel(AccessChannel.LOCAL);
        /**
         * 设置为您从阿里云控制台获取的接入点信息，类似“http://MQ_INST_XXXX.aliyuncs.com:80”。
         */
        consumer.setNamesrvAddr("127.0.0.1:9876");
        try {
            consumer.subscribe("topicSequence", "tagSequence");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
               try {
                   for (MessageExt msg : msgs) {
                       byte[] body = msg.getBody();
                       System.out.println("msg"+ JSONObject.toJSONString(msg)+"消息体：" +new java.lang.String(body));
                   }
               }catch (Exception e){
                   e.getStackTrace();
               }
                return ConsumeOrderlyStatus.SUCCESS;// 消费失败则挂起重试返回：ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT
            }
        });
        try {
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}
